#include "app_pool.h"
#include "mqueue.h"
#include "pthread.h"
#include "log.h"
#include "errno.h"
#include "stdlib.h"

// 申明一个全局的队列
static mqd_t mqd;

// 申明一个全局的线程池
static pthread_t threads[THREAD_NUM];

/**
 * 每个线程要持续做的事情
 * 1. 循环从队列中获取任务
 * 2. 执行任务
 */
static void *app_pool_worker_func(void *arg)
{
    log_info("app_pool thread start");
    while (1)
    {
        Task task;
        // 1. 从队列中获取任务
        // 接受消息 参数 1. 消息队列的描述符 2. 接受消息的缓冲区指针 3. 消息缓冲区大小 4. 消息优先级
        mq_receive(mqd, (char *)&task, sizeof(Task), 0);
        // 2. 执行任务
        task.task_func(task.arg);
    }
}

/**
 *  初始化线程 池
 *  1.创建队列
 *  2.创建n线程数组
 */
int app_pool_init()
{
    // 0. 清理旧的队列文件
    mq_unlink(MQ_NAME);
    // 1. 创建队列
    // 队列的属性
    struct mq_attr attr;
    // 队列中可以存储的最大消息数
    attr.mq_maxmsg = MQ_MAXSIZE;
    // 每个消息的最大字节数
    attr.mq_msgsize = sizeof(Task);
    // 参数 1. 消息队列的名称，必须以"/"开头 2. 消息队列的打开模式 3. 消息队列的属性 
    // 返回值：-1失败，成功：返回一个消息队列的描述符
    mqd = mq_open(MQ_NAME, O_CREAT | O_RDWR, 0666, &attr);
    if (mqd == -1) 
    {
        perror("mq_open failed");
        return -1;
    }
    // 2. 初始化线程数组
    for (int i = 0; i < THREAD_NUM; i++)
    {   
        // 启动一个新的线程，让该线程执行一个指定的函数
        // 创建线程 参数 1. 线程的id 2. 线程的属性 3. 函数指针，线程要执行的函数 4. 传递给函数的参数
        int ret = pthread_create(&threads[i], NULL, app_pool_worker_func, NULL);
        if (ret != 0)
        {
            perror("pthread_create failed");
            return -1;

        }
    }
    return 0;
}

/**
 * 向线程池添加任务
 * 1.把函数和参数封装为task结构体
 * 2.把task结构体放入队列中
 */
int app_pool_add_task(void *(*task_func)(void *arg), void *arg)
{
    // 1.把函数和参数封装为task结构体
    Task *task = (Task *)malloc(sizeof(Task));
    task->task_func = task_func;
    task->arg = arg;
    // 2. 把task结构体放入队列中
    // 参数 1. 消息队列的描述符 2. 指向要发送的消息的指针 3. 消息的大小 4. 消息的优先级
    // 返回值：-1失败，成功：返回0
    int ret = mq_send(mqd, (char *)task, sizeof(Task), 0);
    if (ret == -1) 
    {
        perror("mq_send failed");
        return -1;
    }
    return 0;
}

/**
 * 销毁线程池
 */
int app_pool_destroy()
{
    for (size_t i = 0; i < THREAD_NUM; i++)
    {
        pthread_cancel(threads[i]);
        pthread_join(threads[i], NULL);
    }
    
    mq_close(mqd);
    mq_unlink(MQ_NAME);
    
}
